package cn._51doit.live.udf;

import cn._51doit.live.pojo.ItemEventCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

public class HotGoodTopNSortFunction extends KeyedProcessFunction<Tuple4<String, String, Long, Long>, ItemEventCount, ItemEventCount> {

    private transient ListState<ItemEventCount> listState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<ItemEventCount> stateDescriptor = new ListStateDescriptor<>("list-state", ItemEventCount.class);
        listState = getRuntimeContext().getListState(stateDescriptor);
    }

    @Override
    public void processElement(ItemEventCount input, Context ctx, Collector<ItemEventCount> out) throws Exception {

        listState.add(input);
        //注册定时器
        ctx.timerService().registerEventTimeTimer(input.windowEnd + 1);

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<ItemEventCount> out) throws Exception {

        ArrayList<ItemEventCount> lst = (ArrayList<ItemEventCount>) listState.get();
        //排序
        lst.sort(new Comparator<ItemEventCount>() {
            @Override
            public int compare(ItemEventCount o1, ItemEventCount o2) {
                return Long.compare(o2.count, o1.count);
            }
        });
        for (int i = 0; i < Math.min(3, lst.size()); i++) {
            ItemEventCount itemEventCount = lst.get(i);
            itemEventCount.order = i + 1;
            //输出数据
            out.collect(itemEventCount);
        }
    }
}
